package org.example.lookup_join;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class FlinkRowTimeLookupJoin {
    public static void main(String[] args) throws Exception {
        // 配置Flink Web UI端口
        Configuration config = new Configuration();
        config.setString("rest.bind-port", "8081"); // 设置Web UI端口为8085

        // 设置流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
        env.setParallelism(1); // 设置并行度为1
        env.enableCheckpointing(5000); // 开启checkpoint，间隔5秒
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        // 创建Paimon目录
        String createCatalogSQL = "CREATE CATALOG paimon WITH (\n" +
                "    'type' = 'paimon',\n" +
                "    'warehouse' = 'file:///tmp/paimon'\n" +
                ");";
        System.out.println("执行SQL: " + createCatalogSQL);
        tableEnv.executeSql(createCatalogSQL);

        // 在Lookup Join之前添加datagen表创建语句
        String createDatagenTableSQL = "CREATE TABLE IF NOT EXISTS `datagen_t_order_table` (\n" +
                "    order_id BIGINT COMMENT '订单ID',\n" +
                "    user_id BIGINT COMMENT '用户ID',\n" +
                "    order_amount DOUBLE COMMENT '订单金额',\n" +
                "    order_date TIMESTAMP(3) COMMENT '订单时间',\n" +
                "    WATERMARK FOR order_date AS order_date - INTERVAL '5' SECOND\n" +
                ") WITH (\n" +
                "    'connector' = 'datagen',\n" +
                "    'rows-per-second' = '1',\n" +
                "    'number-of-rows' = '100',\n" +
                "    'fields.user_id.min' = '1',\n" +
                "    'fields.user_id.max' = '100',\n" +
                "    'fields.order_amount.min' = '10',\n" +
                "    'fields.order_amount.max' = '1000'\n" +
                ");";
        System.out.println("执行SQL: " + createDatagenTableSQL);
        tableEnv.executeSql(createDatagenTableSQL);

        // 执行Lookup Join查询，使用event time进行时间窗口计算
        String lookupJoinSQL = "SELECT \n" +
                "    o.order_id, \n" +
                "    o.user_id, \n" +
                "    u.user_name, \n" +
                "    o.order_amount, \n" +
                "    o.order_date \n" +
                "FROM `datagen_t_order_table` AS o \n" +
                "JOIN `paimon`.`default`.`t_user_table_with_eventtime` FOR SYSTEM_TIME AS OF o.order_date AS u \n" +
                "ON o.user_id = u.user_id";
        System.out.println("执行SQL: " + lookupJoinSQL);
        tableEnv.executeSql(lookupJoinSQL).print();

        // 统计每个用户名对应的订单数、订单金额、最近下单时间，基于event time进行窗口计算
        String statsSQL = "SELECT \n" +
                "    user_name, \n" +
                "    COUNT(order_id) AS order_count, \n" +
                "    SUM(order_amount) AS total_amount, \n" +
                "    MAX(order_date) AS last_order_time \n" +
                "FROM (" + lookupJoinSQL + ") \n" +
                "GROUP BY user_name, TUMBLE(order_date, INTERVAL '5' SECOND);";
        System.out.println("执行SQL: " + statsSQL);
        tableEnv.executeSql(statsSQL).print();
    }
} 